package com.jiayuan.cn.energy.test.ok;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

@Slf4j
public class StreamingJobSql {


    /**
     * Skeleton for a Flink Streaming Job.
     *
     * <p>For a tutorial how to write a Flink streaming application, check the
     * tutorials and examples on the <a href="http://flink.apache.org/docs/stable/">Flink Website</a>.
     *
     * <p>To package your appliation into a JAR file for execution, run
     * 'mvn clean package' on the command line.
     *
     * <p>If you change the name of the main class (with the public static void main(String[] args))
     * method, change the respective entry in the POM.xml file (simply search for 'mainClass').
     */
    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Configuration configuration = new Configuration();
        configuration.setString(RestOptions.BIND_PORT, "8081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        env.setParallelism(1);
        //接收消息并将数据转化为流
        DataStream<String> stream = env.addSource(new MqttConsumer());
//        DataStream<Tuple2<String, String>> dataStream = stream.flatMap(new FlatMapFunction<String, Tuple2<String, String>>() {
//            @Override
//            public void flatMap(String value, Collector<Tuple2<String, String>> out) throws Exception {
//                log.info("value: " + value);
//                out.collect(Tuple2.of("name", value));
//            }
//        }).setParallelism(1);//线程并行数量
//        dataStream.print();
//        env.execute();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String createSinkTable = """
                CREATE TABLE test3 (
                        data STRING
                    ) WITH (
                        'connector'  = 'jdbc',
                        'url'        = 'jdbc:mysql://192.168.110.88:3306/helm?useUnicode=true&characterEncoding=utf-8&useSSL=false',
                        'table-name' = 'test3',
                        'driver'     = 'com.mysql.cj.jdbc.Driver',
                        'username'   = 'helm',
                        'password'   = 'db-999'
                    )
                """;
        tableEnv.executeSql(createSinkTable);
        Table table = tableEnv.fromDataStream(stream).as("data");
        tableEnv.createTemporaryView("model", table);

//        Table resultTable = tableEnv.sqlQuery(
//                "SELECT data FROM model");
//        DataStream<Row> resultStream = tableEnv.toChangelogStream(resultTable);
//        resultStream.print();
//        env.execute();

        String query = "INSERT INTO test3 SELECT data as data  FROM model ";
        tableEnv.executeSql(query);

    }
}